// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jshin

#pragma once


#include <glog/logging.h>

#include <boost/asio/post.hpp>

#include "coin2/exchange/base/feed/distributor.h"
#include "coin2/exchange/base/feed/subscriber.h"
#include "coin2/exchange/base/feed/counter/counter.h"
#include "coin2/exchange/base/feed/replay/processor.h"
#include "coin2/exchange/base/feed/replay/book.h"
#include "coin2/exchange/base/feed/replay/kline.h"
#include "coin2/exchange/base/feed/replay/trading_data.h"

namespace coin2::exchange::base::feed::replay {

using coin2::exchange::base::feed::counter::FeedCounter;
using coin2::exchange::base::feed::replay::ReplayProcessor;
using coin2::exchange::base::feed::replay::ReplayTelemetryProcessor;
using coin2::exchange::base::executor::LifecycleTimestamp;
using coin2::exchange::base::feed_util::Index;

class FeedRelayDistributor : public FeedDistributor {
 public:
  FeedRelayDistributor(
    const std::string& system_name,
    const MarketExchangeApi& mea,
    boost::asio::io_context* replay_ioc,
    int book_depth,
    FeedCounter* counter)
    : replay_ioc_(replay_ioc),
      replay_processor_(system_name, mea, counter),
      book_builder_(book_depth),
      feed_counter_(counter) {
  }

  virtual void AddProduct(const IProduct& product, IFeedSubscriber* subscriber) {
    replay_processor_.AddProduct(product, subscriber);
    FeedDistributor::AddProduct(product, subscriber);
  }

  virtual void RemoveProduct(const std::vector<const IProduct*>& products) {
    replay_processor_.RemoveProduct(products);
    FeedDistributor::RemoveProduct(products);
  }

  template<typename T>
  inline void onMessage(int64_t timestamp, std::string_view native_symbol, const T& data) {
    std::shared_ptr<ReplayMessage<T>> message = std::shared_ptr<ReplayMessage<T>>( 
        new ReplayMessage<T>{
            timestamp, std::move(std::string(native_symbol)), data});
    boost::asio::post(
        *replay_ioc_,
        std::bind(
            &ReplayProcessor::onSharedMessage<T>,
            &replay_processor_,
            message));
    if (feed_counter_) {
      feed_counter_->PublishOneMsg();
    }
  }

  virtual void onTradeFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& trade = upd.trade();
    onMessage<Trade>(upd.timestamp(), native_symbol, trade);
  }

  virtual void onBookFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& book = upd.book();
    BookEvent event = book_builder_.ToEvent(book);
    onMessage<BookEvent>(upd.timestamp(), native_symbol, event);
  }

  virtual void onLiquidationFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& liquidation_order = upd.liquidation_order();
    onMessage<LiquidationOrder>(upd.timestamp(), native_symbol, liquidation_order);
  }

  virtual void onOpenInterestFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& open_interest = upd.open_interest();
    onMessage<OpenInterest>(upd.timestamp(), native_symbol, open_interest);
  }

  virtual void onMarkPriceFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& mark_price = upd.mark_price();
    onMessage<MarkPrice>(upd.timestamp(), native_symbol, mark_price);
  }

  virtual void onIndexFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& index = upd.index();
    onMessage<Index>(upd.timestamp(), native_symbol, index);
  }

  virtual void onFundingRateFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& funding_rate = upd.funding_rate();
    onMessage<FundingRate>(upd.timestamp(), native_symbol, funding_rate);
  }

  virtual void onIvGreeksFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& iv_greeks = upd.iv_greeks();
    onMessage<IvGreeks>(upd.timestamp(), native_symbol, iv_greeks);
  }

  virtual void onNavFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& nav = upd.nav();
    onMessage<Nav>(upd.timestamp(), native_symbol, nav);
  }

  virtual void onKlineFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& klines = upd.klines();
    auto event = kline_builder_.ToEvent(native_symbol, klines);
    if (!event.klines.empty()) {
      onMessage<KlinesEvent>(upd.timestamp(), native_symbol, event);
    }
  }

  virtual void onWalletFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& wallet = upd.wallet();
    onMessage<Wallet>(upd.timestamp(), native_symbol, wallet);
  }

  virtual void onMarketAnalysisFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& market_analysis = upd.market_analysis();
    onMessage<MarketAnalysis>(upd.timestamp(), native_symbol, market_analysis);
  }

  virtual void onTopLongShortAccountRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& ratio = upd.top_long_short_account_ratio();
    auto event = top_account_builder_.ToEvent(native_symbol, ratio);
    if (!event.data.empty()) {
      onMessage<TradingDataEvent<TopLongShortAccountRatio>>(upd.timestamp(), native_symbol, event);
    }
  }

  virtual void onTopLongShortPositionRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& ratio = upd.top_long_short_position_ratio();
    auto event = top_position_builder_.ToEvent(native_symbol, ratio);
    if (!event.data.empty()) {
      onMessage<TradingDataEvent<TopLongShortPositionRatio>>(upd.timestamp(), native_symbol, event);
    }
  }

  virtual void onGlobalLongShortAccountRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& ratio = upd.global_long_short_account_ratio();
    auto event = global_account_builder_.ToEvent(native_symbol, ratio);
    if (!event.data.empty()) {
      onMessage<TradingDataEvent<GlobalLongShortAccountRatio>>(upd.timestamp(), native_symbol, event);
    }
  }

  virtual void onTakerBuySellRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& ratio = upd.taker_buy_sell_ratio();
    auto event = taker_volume_builder_.ToEvent(native_symbol, ratio);
    if (!event.data.empty()) {
      onMessage<TradingDataEvent<TakerBuySellRatio>>(upd.timestamp(), native_symbol, event);
    }
  }

  virtual void onOpenInterestHistFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& open_interest_hist = upd.open_interest_hist();
    auto event = open_interest_hist_builder_.ToEvent(native_symbol, open_interest_hist);
    if (!event.data.empty()) {
      onMessage<TradingDataEvent<OpenInterest>>(upd.timestamp(), native_symbol, event);
    }
  }

 protected:
  boost::asio::io_context* replay_ioc_ = nullptr;
  ReplayProcessor replay_processor_;

  BookEventBuilder book_builder_;
  KlineEventBuilder kline_builder_;
  TradingDataEventBuilder top_account_builder_;
  TradingDataEventBuilder top_position_builder_;
  TradingDataEventBuilder global_account_builder_;
  TradingDataEventBuilder taker_volume_builder_;
  TradingDataEventBuilder open_interest_hist_builder_;
  FeedCounter* feed_counter_ = nullptr;
};

class FeedRelayTelemetryDistributor : public FeedDistributor {
 public:
  FeedRelayTelemetryDistributor(
    const std::string& system_name,
    const MarketExchangeApi& mea,
    boost::asio::io_context* replay_ioc,
    int book_depth,
    FeedCounter* counter)
    : replay_ioc_(replay_ioc),
      replay_telemetry_processor_(system_name, mea, counter),
      book_builder_(book_depth),
      feed_counter_(counter) {
  }

  virtual void AddProduct(const IProduct& product, IFeedSubscriber* subscriber) {
    replay_telemetry_processor_.AddProduct(product, subscriber);
    FeedDistributor::AddProduct(product, subscriber);
  }

  virtual void RemoveProduct(const std::vector<const IProduct*>& products) {
    replay_telemetry_processor_.RemoveProduct(products);
    FeedDistributor::RemoveProduct(products);
  }

  template<typename T>
  inline void onMessage(const LifecycleTimestamp& lts, int64_t timestamp, std::string_view native_symbol, const T& data) {
    auto message = std::shared_ptr<ReplayLifecycleMessage<T>>( 
        new ReplayLifecycleMessage<T>{
            lts, timestamp, std::move(std::string(native_symbol)), data});
    // origin main_rx is paser_rx in parser ioc mode
    message->lts.set_parser_rx_timestamp(message->lts.main_rx_timestamp());
    message->lts.set_parser_tx_timestamp(GetCurrentTimestamp());
    boost::asio::post(
        *replay_ioc_,
        std::bind(
            &ReplayTelemetryProcessor::onSharedMessage<T>,
            &replay_telemetry_processor_,
            message));
    if (feed_counter_) {
      feed_counter_->PublishOneMsg();
    }
  }

  virtual void onTradeFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& trade = upd.trade();
    onMessage<Trade>(upd.ts(), upd.timestamp(), native_symbol, trade);
  }

  virtual void onBookFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& book = upd.book();
    auto event = book_builder_.ToEvent(book);
    onMessage<BookEvent>(upd.ts(), upd.timestamp(), native_symbol, event);
  }

  virtual void onLiquidationFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& liquidation_order = upd.liquidation_order();
    onMessage<LiquidationOrder>(upd.ts(), upd.timestamp(), native_symbol, liquidation_order);
  }

  virtual void onOpenInterestFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& open_interest = upd.open_interest();
    onMessage<OpenInterest>(upd.ts(), upd.timestamp(), native_symbol, open_interest);
  }

  virtual void onMarkPriceFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& mark_price = upd.mark_price();
    onMessage<MarkPrice>(upd.ts(), upd.timestamp(), native_symbol, mark_price);
  }

  virtual void onIndexFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& index = upd.index();
    onMessage<Index>(upd.ts(), upd.timestamp(), native_symbol, index);
  }

  virtual void onFundingRateFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& funding_rate = upd.funding_rate();
    onMessage<FundingRate>(upd.ts(), upd.timestamp(), native_symbol, funding_rate);
  }

  virtual void onIvGreeksFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& iv_greeks = upd.iv_greeks();
    onMessage<IvGreeks>(upd.ts(), upd.timestamp(), native_symbol, iv_greeks);
  }

  virtual void onNavFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& nav = upd.nav();
    onMessage<Nav>(upd.ts(), upd.timestamp(), native_symbol, nav);
  }

  virtual void onKlineFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& klines = upd.klines();
    auto event = kline_builder_.ToEvent(native_symbol, klines);
    if (!event.klines.empty()) {
      onMessage<KlinesEvent>(upd.ts(), upd.timestamp(), native_symbol, event);
    }
  }

  virtual void onWalletFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& wallet = upd.wallet();
    onMessage<Wallet>(upd.ts(), upd.timestamp(), native_symbol, wallet);
  }

  virtual void onMarketAnalysisFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& market_analysis = upd.market_analysis();
    onMessage<MarketAnalysis>(upd.ts(), upd.timestamp(), native_symbol, market_analysis);
  }

  virtual void onTopLongShortAccountRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    const auto& ratio = upd.top_long_short_account_ratio();
    auto event = top_account_builder_.ToEvent(native_symbol, ratio);
    if (!event.data.empty()) {
      onMessage<TradingDataEvent<TopLongShortAccountRatio>>(upd.ts(), upd.timestamp(), native_symbol, event);
    }
  }

  virtual void onTopLongShortPositionRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& ratio = upd.top_long_short_position_ratio();
    auto event = top_position_builder_.ToEvent(native_symbol, ratio);
    if (!event.data.empty()) {
      onMessage<TradingDataEvent<TopLongShortPositionRatio>>(upd.ts(), upd.timestamp(), native_symbol, event);
    }
  }

  virtual void onGlobalLongShortAccountRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& ratio = upd.global_long_short_account_ratio();
    auto event = global_account_builder_.ToEvent(native_symbol, ratio);
    if (!event.data.empty()) {
      onMessage<TradingDataEvent<GlobalLongShortAccountRatio>>(upd.ts(), upd.timestamp(), native_symbol, event);
    }
  }

  virtual void onTakerBuySellRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& ratio = upd.taker_buy_sell_ratio();
    auto event = taker_volume_builder_.ToEvent(native_symbol, ratio);
    if (!event.data.empty()) {
      onMessage<TradingDataEvent<TakerBuySellRatio>>(upd.ts(), upd.timestamp(), native_symbol, event);
    }
  }

  virtual void onOpenInterestHistFeed(std::string_view native_symbol, const FeedUpdate& upd) override {
    auto& open_interest_hist = upd.open_interest_hist();
    auto event = open_interest_hist_builder_.ToEvent(native_symbol, open_interest_hist);
    if (!event.data.empty()) {
      onMessage<TradingDataEvent<OpenInterest>>(upd.ts(), upd.timestamp(), native_symbol, event);
    }
  }

 protected:
  boost::asio::io_context* replay_ioc_ = nullptr;
  ReplayTelemetryProcessor replay_telemetry_processor_;

  BookEventBuilder book_builder_;
  KlineEventBuilder kline_builder_;
  TradingDataEventBuilder top_account_builder_;
  TradingDataEventBuilder top_position_builder_;
  TradingDataEventBuilder global_account_builder_;
  TradingDataEventBuilder taker_volume_builder_;
  TradingDataEventBuilder open_interest_hist_builder_;
  FeedCounter* feed_counter_ = nullptr;
};

}  // namespace coin2::exchange::base::feed::replay
